package com.adley.rocket.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumerConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumerConcurrentlyStatus;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Confuration;
import org.springframework.core.env.Environment;
import org.springframework.web.context.WebApplicationContext;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ExecutorService;

@Slf4j
@Configuration
public class RocketmqAutoConfiguration {
    @Autowired
    private WebApplicationContext wac;
    @Autowired
    private Environment env;

    @Bean
    public DefaultMQProducer defaultProducer() throws MQClientException {
        DefualtMQProducer producer = new DefaultMQProducer(env.getProperty("rocketmq.producer.group"));
        producer.setNamesrvAddr(env.getProperty("rocketmq.name-srv-addr"));
        producer.setInstanceName(env.getProperty("rocketmq.producer.name"));
        producer.setVipChannelEnabled(true);
        producer.setAsyncSenderExecutor(wac.getBean(ExecutorService.class));

        producer.start();
        log.info("Rocketmq started");
        return producer;
    }
    
    @Bean
    public DefaultMQPushConsumer pushConsumer() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(env.getProperty("rocketmq.consumer.group"));
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setNamesrvAddr(env.getProperty("rocketmq.name-srv-addr"));
        consumer.setInstanceName(env.getProperty("rocketmq.consumer.name"));
        consumer.setAsyncSenderExecutor(wac.getBean(ExecutorService.class));

        List<String> subscribeList = Arrays.asList(env.getProperty("rocketmq.consumer.topics").split(","));
        for (String s : subscribeList) {
            consumer.subscribe(s.split(":")[0],s.split(":")[1]);
        }

        consumer.registerMessageListenter((List<MessageExt) msgs, ConsumeConcurrentlyContext context) -> {
            try {
                MessageExt msg = msgs.get(0);
                log.info("New message = {}", msg);
            } catch(Exception e) {
                log.error("{}", e);
                retur ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
            retur ConsumeConcurrentlyStatus.RECONSUME_SUCCESS;
        });

        new Thread(() -> {
            try {
                Thread.sleep(5000); //延迟启动，目的是等待spring事件监听相关程序初始化完成；否则，会出现消息到达时上面消费事件还未注册完成，导致消息丢失
                try {
                    consumer.start();
                } catch (Exception e) {
                    log.error("Rocketmq consumer start failed...");    
                }
            } catch (Exception e) {
                log.error("Rocketmq consumer start failed...");
            }
        }).start();
        return consumer;
    }
}
